package xin.alum.aim.cluster;

import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import xin.alum.aim.constant.AimConstant;
import xin.alum.aim.constant.Constants;
import xin.alum.aim.groups.ClusterFactory;
import xin.alum.aim.groups.Session;
import xin.alum.aim.model.Transportable;

import java.util.Comparator;
import java.util.concurrent.PriorityBlockingQueue;

public abstract class ClusterPusher implements ClusterFactory {

    protected final InternalLogger logger = InternalLoggerFactory.getInstance(this.getClass());
    private final Comparator<Transportable> comparator = (p1, p2) -> p1.getPriority() > p2.getPriority() ? p1.getPriority() : p2.getPriority();
    protected PriorityBlockingQueue<Transportable> queue = new PriorityBlockingQueue<>(ClusterFactory.QUEUE_CAPACITY, comparator);

    protected abstract <T> boolean channel(String channel, T d);

    protected void pushQueue(){
        if (queue.size() > 10) {
            logger.warn("集群消息出现积压:{}/{}", queue.size(), ClusterFactory.QUEUE_CAPACITY);
            //queue.removeIf(w->w.getData().getPriority()<=0);
        }
        Transportable m;
        while (queue.iterator().hasNext()) {
            m = queue.poll();
            if (m == null) break;
            switch (m.getTag()) {
                case AimConstant.DATA_TYPE_REPLY:
                    channel(Constants.EVENT_MESSAGE_INNER_QUEUE, m);
                    break;
                case AimConstant.DATA_TYPE_MESSAGE:
                    channel(Constants.PUSH_MESSAGE_INNER_QUEUE,  m);
                    break;
                default:
                    logger.error("无效Type类型：{}", m);
                    break;
            }
        }
    }

    @Override
    public void kick(Session session) {
        channel(Constants.BIND_MESSAGE_INNER_QUEUE, session);
    }

    @Override
    public void push(String recipient, Transportable data) {
        data.setRecipient(recipient);
        queue.put(data);
        pushQueue();
    }
}
